// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <nebula/util/windows_compatibility.h>  // IWYU pragma: keep

// sys/mman.h not present in Visual Studio or Cygwin
#ifdef _WIN32
#ifndef NOMINMAX
#define NOMINMAX
#endif
#include <nebula/io/mman.h>
#undef Realloc
#undef Free
#else

#include <fcntl.h>
#include <sys/mman.h>
#include <unistd.h>  // IWYU pragma: keep

#endif

#include <algorithm>
#include <atomic>
#include <cerrno>
#include <cstdint>
#include <cstring>
#include <limits>
#include <memory>
#include <mutex>
#include <sstream>
#include <string>
#include <utility>

// ----------------------------------------------------------------------
// Other Nebula includes

#include <nebula/io/file.h>
#include <nebula/io/interfaces.h>
#include <nebula/io/util_internal.h>

#include <nebula/core/buffer.h>
#include <nebula/core/memory_pool.h>
#include <turbo/utility/status.h>
#include <nebula/future/future.h>
#include <turbo/files/io_util.h>
#include <turbo/log/logging.h>

namespace nebula {

    namespace io {

        class OSFile {
        public:
            // Note: only one of the open* methods below may be called on a given instance

            turbo::Status OpenWritable(const std::string &path, bool truncate, bool append,
                                bool write_only) {
                TURBO_RETURN_NOT_OK(SetFileName(path));

                TURBO_MOVE_OR_RAISE(fd_, turbo::file_open_writable(file_name_, write_only,
                                                                                 truncate, append));
                mode_ = write_only ? FileMode::WRITE : FileMode::READWRITE;

                if (!truncate) {
                    TURBO_MOVE_OR_RAISE(size_, turbo::file_get_size(fd_.fd()));
                } else {
                    size_ = 0;
                }
                return turbo::OkStatus();
            }

            // This is different from OpenWritable(string, ...) in that it doesn't
            // truncate nor mandate a seekable file
            turbo::Status OpenWritable(int fd) {
                auto result = turbo::file_get_size(fd);
                if (result.ok()) {
                    size_ = *result;
                } else {
                    // Non-seekable file
                    size_ = -1;
                }
                TURBO_RETURN_NOT_OK(SetFileName(fd));
                mode_ = FileMode::WRITE;
                fd_ = turbo::FileDescriptor(fd);
                return turbo::OkStatus();
            }

            turbo::Status OpenReadable(const std::string &path) {
                TURBO_RETURN_NOT_OK(SetFileName(path));

                TURBO_MOVE_OR_RAISE(fd_, turbo::file_open_readable(file_name_));
                TURBO_MOVE_OR_RAISE(size_, turbo::file_get_size(fd_.fd()));

                mode_ = FileMode::READ;
                return turbo::OkStatus();
            }

            turbo::Status OpenReadable(int fd) {
                TURBO_MOVE_OR_RAISE(size_, turbo::file_get_size(fd));
                TURBO_RETURN_NOT_OK(SetFileName(fd));
                mode_ = FileMode::READ;
                fd_ = turbo::FileDescriptor(fd);
                return turbo::OkStatus();
            }

            turbo::Status CheckClosed() const {
                if (fd_.closed()) {
                    return turbo::invalid_argument_error("Invalid operation on closed file");
                }
                return turbo::OkStatus();
            }

            turbo::Status close() { return fd_.close(); }

            turbo::Result<int64_t> read(int64_t nbytes, void *out) {
                TURBO_RETURN_NOT_OK(CheckClosed());
                TURBO_RETURN_NOT_OK(CheckPositioned());
                return turbo::file_read(fd_.fd(), reinterpret_cast<uint8_t *>(out), nbytes);
            }

            turbo::Result<int64_t> read_at(int64_t position, int64_t nbytes, void *out) {
                TURBO_RETURN_NOT_OK(CheckClosed());
                TURBO_RETURN_NOT_OK(internal::ValidateRange(position, nbytes));
                // read_at() leaves the file position undefined, so require that we seek
                // before calling read() or write().
                need_seeking_.store(true);
                return turbo::file_read_at(fd_.fd(), reinterpret_cast<uint8_t *>(out),
                                                      position, nbytes);
            }

            turbo::Status Seek(int64_t pos) {
                TURBO_RETURN_NOT_OK(CheckClosed());
                if (pos < 0) {
                    return turbo::invalid_argument_error("Invalid position");
                }
                turbo::Status st = turbo::file_seek(fd_.fd(), pos);
                if (st.ok()) {
                    need_seeking_.store(false);
                }
                return st;
            }

            turbo::Result<int64_t> tell() const {
                TURBO_RETURN_NOT_OK(CheckClosed());
                return turbo::file_tell(fd_.fd());
            }

            turbo::Status write(const void *data, int64_t length) {
                TURBO_RETURN_NOT_OK(CheckClosed());

                std::lock_guard<std::mutex> guard(lock_);
                TURBO_RETURN_NOT_OK(CheckPositioned());
                if (length < 0) {
                    return turbo::io_error("Length must be non-negative");
                }
                return turbo::file_write(fd_.fd(), reinterpret_cast<const uint8_t *>(data),
                                                     length);
            }

            int fd() const { return fd_.fd(); }

            bool is_open() const { return !fd_.closed(); }

            int64_t size() const { return size_; }

            FileMode::type mode() const { return mode_; }

            std::mutex &lock() { return lock_; }

        protected:
            turbo::Status SetFileName(const std::string &file_name) {
                return turbo::FilePath::from_string(file_name).try_value(&file_name_);
            }

            turbo::Status SetFileName(int fd) {
                std::stringstream ss;
                ss << "<fd " << fd << ">";
                return SetFileName(ss.str());
            }

            turbo::Status CheckPositioned() {
                if (need_seeking_.load()) {
                    return turbo::invalid_argument_error(
                            "Need seeking after read_at() before "
                            "calling implicitly-positioned operation");
                }
                return turbo::OkStatus();
            }

            turbo::FilePath file_name_;

            std::mutex lock_;
            turbo::FileDescriptor fd_;
            FileMode::type mode_;
            int64_t size_{-1};
            // Whether read_at made the file position non-deterministic.
            std::atomic<bool> need_seeking_{false};
        };

        // ----------------------------------------------------------------------
        // ReadableFile implementation

        class ReadableFile::ReadableFileImpl : public OSFile {
        public:
            explicit ReadableFileImpl(MemoryPool *pool) : OSFile(), pool_(pool) {}

            turbo::Status open(const std::string &path) { return OpenReadable(path); }

            turbo::Status open(int fd) { return OpenReadable(fd); }

            turbo::Result<std::shared_ptr<Buffer>> read_buffer(int64_t nbytes) {
                TURBO_MOVE_OR_RAISE(auto buffer, allocate_resizable_buffer(nbytes, pool_));

                TURBO_MOVE_OR_RAISE(int64_t bytes_read, read(nbytes, buffer->mutable_data()));
                if (bytes_read < nbytes) {
                    TURBO_RETURN_NOT_OK(buffer->resize(bytes_read));
                    buffer->zero_padding();
                }
                // R build with openSUSE155 requires an explicit shared_ptr construction
                return std::shared_ptr<Buffer>(std::move(buffer));
            }

            turbo::Result<std::shared_ptr<Buffer>> ReadBufferAt(int64_t position, int64_t nbytes) {
                TURBO_MOVE_OR_RAISE(auto buffer, allocate_resizable_buffer(nbytes, pool_));

                TURBO_MOVE_OR_RAISE(int64_t bytes_read,
                                       read_at(position, nbytes, buffer->mutable_data()));
                if (bytes_read < nbytes) {
                    TURBO_RETURN_NOT_OK(buffer->resize(bytes_read));
                    buffer->zero_padding();
                }
                // R build with openSUSE155 requires an explicit shared_ptr construction
                return std::shared_ptr<Buffer>(std::move(buffer));
            }

            turbo::Status will_need(const std::vector<ReadRange> &ranges) {
                auto report_error = [](int errnum, const char *msg) -> turbo::Status {
                    if (errnum == EBADF || errnum == EINVAL) {
                        // These are logic errors, so raise them
                        return turbo::io_error_with_errno_payload(errnum, msg);
                    }
#ifndef NDEBUG
                    // Other errors may be encountered if the target device or filesystem
                    // does not support fadvise advisory (for example, macOS can return
                    // ENOTTY on macOS: ARROW-13983).  Log the error for diagnosis
                    // on debug builds, but avoid bothering the user otherwise.
                    KLOG(WARNING) << turbo::io_error_with_errno_payload(errnum, msg).to_string();
#else
                    TURBO_UNUSED(msg);
#endif
                    return turbo::OkStatus();
                };
                TURBO_RETURN_NOT_OK(CheckClosed());
                for (const auto &range: ranges) {
                    TURBO_RETURN_NOT_OK(internal::ValidateRange(range.offset, range.length));
#if defined(POSIX_FADV_WILLNEED)
                    int ret = posix_fadvise(fd_.fd(), range.offset, range.length, POSIX_FADV_WILLNEED);
                    if (ret) {
                        TURBO_RETURN_NOT_OK(report_error(ret, "posix_fadvise failed"));
                    }
#elif defined(F_RDADVISE)  // macOS, BSD?
                    struct {
                      off_t ra_offset;
                      int ra_count;
                    } radvisory{range.offset, static_cast<int>(range.length)};
                    if (radvisory.ra_count > 0 && fcntl(fd_.fd(), F_RDADVISE, &radvisory) == -1) {
                      TURBO_RETURN_NOT_OK(report_error(errno, "fcntl(fd, F_RDADVISE, ...) failed"));
                    }
#else
                    TURBO_UNUSED(report_error);
#endif
                }
                return turbo::OkStatus();
            }

        private:
            MemoryPool *pool_;
        };

        ReadableFile::ReadableFile(MemoryPool *pool) { impl_.reset(new ReadableFileImpl(pool)); }

        ReadableFile::~ReadableFile() { internal::CloseFromDestructor(this); }

        turbo::Result<std::shared_ptr<ReadableFile>> ReadableFile::open(const std::string &path,
                                                                 MemoryPool *pool) {
            auto file = std::shared_ptr<ReadableFile>(new ReadableFile(pool));
            TURBO_RETURN_NOT_OK(file->impl_->open(path));
            return file;
        }

        turbo::Result<std::shared_ptr<ReadableFile>> ReadableFile::open(int fd, MemoryPool *pool) {
            auto file = std::shared_ptr<ReadableFile>(new ReadableFile(pool));
            TURBO_RETURN_NOT_OK(file->impl_->open(fd));
            return file;
        }

        turbo::Status ReadableFile::DoClose() { return impl_->close(); }

        bool ReadableFile::closed() const { return !impl_->is_open(); }

        turbo::Status ReadableFile::will_need(const std::vector<ReadRange> &ranges) {
            return impl_->will_need(ranges);
        }

        turbo::Result<int64_t> ReadableFile::DoTell() const { return impl_->tell(); }

        turbo::Result<int64_t> ReadableFile::DoRead(int64_t nbytes, void *out) {
            return impl_->read(nbytes, out);
        }

        turbo::Result<int64_t> ReadableFile::DoReadAt(int64_t position, int64_t nbytes, void *out) {
            return impl_->read_at(position, nbytes, out);
        }

        turbo::Result<std::shared_ptr<Buffer>> ReadableFile::DoReadAt(int64_t position, int64_t nbytes) {
            return impl_->ReadBufferAt(position, nbytes);
        }

        turbo::Result<std::shared_ptr<Buffer>> ReadableFile::DoRead(int64_t nbytes) {
            return impl_->read_buffer(nbytes);
        }

        turbo::Result<int64_t> ReadableFile::DoGetSize() { return impl_->size(); }

        turbo::Status ReadableFile::DoSeek(int64_t pos) { return impl_->Seek(pos); }

        int ReadableFile::file_descriptor() const { return impl_->fd(); }

        // ----------------------------------------------------------------------
        // FileOutputStream

        class FileOutputStream::FileOutputStreamImpl : public OSFile {
        public:
            turbo::Status open(const std::string &path, bool append) {
                const bool truncate = !append;
                return OpenWritable(path, truncate, append, true /* write_only */);
            }

            turbo::Status open(int fd) { return OpenWritable(fd); }
        };

        FileOutputStream::FileOutputStream() { impl_.reset(new FileOutputStreamImpl()); }

        FileOutputStream::~FileOutputStream() { internal::CloseFromDestructor(this); }

        turbo::Result<std::shared_ptr<FileOutputStream>> FileOutputStream::open(const std::string &path,
                                                                         bool append) {
            auto stream = std::shared_ptr<FileOutputStream>(new FileOutputStream());
            TURBO_RETURN_NOT_OK(stream->impl_->open(path, append));
            return stream;
        }

        turbo::Result<std::shared_ptr<FileOutputStream>> FileOutputStream::open(int fd) {
            auto stream = std::shared_ptr<FileOutputStream>(new FileOutputStream());
            TURBO_RETURN_NOT_OK(stream->impl_->open(fd));
            return stream;
        }

        turbo::Status FileOutputStream::close() { return impl_->close(); }

        bool FileOutputStream::closed() const { return !impl_->is_open(); }

        turbo::Result<int64_t> FileOutputStream::tell() const { return impl_->tell(); }

        turbo::Status FileOutputStream::write(const void *data, int64_t length) {
            return impl_->write(data, length);
        }

        int FileOutputStream::file_descriptor() const { return impl_->fd(); }

        // ----------------------------------------------------------------------
        // Implement MemoryMappedFile

        class MemoryMappedFile::MemoryMap
                : public std::enable_shared_from_this<MemoryMappedFile::MemoryMap> {
        public:
            // An object representing the entire memory-mapped region.
            // It can be sliced in order to return individual subregions, which
            // will then keep the original region alive as long as necessary.
            class Region : public Buffer {
            public:
                Region(std::shared_ptr<MemoryMappedFile::MemoryMap> memory_map, uint8_t *data,
                       int64_t size)
                        : Buffer(data, size) {
                    is_mutable_ = memory_map->writable();
                }

                ~Region() {
                    if (data_ != nullptr) {
#ifndef __EMSCRIPTEN__
                        int result = munmap(data(), static_cast<size_t>(size_));
                        // emscripten erroneously reports failures in munmap
                        // https://github.com/emscripten-core/emscripten/issues/20459
                        KCHECK_EQ(result, 0) << "munmap failed";
#else
                        munmap(data(), static_cast<size_t>(size_));
#endif
                    }
                }

                // For convenience
                uint8_t *data() { return const_cast<uint8_t *>(data_); }

                void detach() { data_ = nullptr; }
            };

            MemoryMap() : file_size_(0), map_len_(0) {}

            ~MemoryMap() { KCHECK_OK(close()); }

            turbo::Status close() {
                if (file_->is_open()) {
                    // Lose our reference to the MemoryMappedRegion, so that munmap()
                    // is called as soon as all buffer exports are released.
                    region_.reset();
                    return file_->close();
                } else {
                    return turbo::OkStatus();
                }
            }

            bool closed() const { return !file_->is_open(); }

            turbo::Status CheckClosed() const {
                if (closed()) {
                    return turbo::invalid_argument_error("Invalid operation on closed file");
                }
                return turbo::OkStatus();
            }

            turbo::Status open(const std::string &path, FileMode::type mode, const int64_t offset = 0,
                        const int64_t length = -1) {
                file_ = std::make_unique<OSFile>();

                if (mode != FileMode::READ) {
                    // Memory mapping has permission failures if PROT_READ not set
                    prot_flags_ = PROT_READ | PROT_WRITE;
                    map_mode_ = MAP_SHARED;
                    constexpr bool append = false;
                    constexpr bool truncate = false;
                    constexpr bool write_only = false;
                    TURBO_RETURN_NOT_OK(file_->OpenWritable(path, truncate, append, write_only));
                } else {
                    prot_flags_ = PROT_READ;
                    map_mode_ = MAP_PRIVATE;  // Changes are not to be committed back to the file
                    TURBO_RETURN_NOT_OK(file_->OpenReadable(path));
                }
                map_len_ = offset_ = 0;

                // Memory mapping fails when file size is 0
                // delay it until the first resize
                if (file_->size() > 0) {
                    TURBO_RETURN_NOT_OK(InitMMap(file_->size(), false, offset, length));
                }

                position_ = 0;

                return turbo::OkStatus();
            }

            // resize the mmap and file to the specified size.
            // resize on memory mapped file region is not supported.
            turbo::Status resize(const int64_t new_size) {
                if (!writable()) {
                    return turbo::io_error("Cannot resize a readonly memory map");
                }
                if (map_len_ != file_size_) {
                    return turbo::io_error("Cannot resize a partial memory map");
                }
                if (region_.use_count() > 1) {
                    // There are buffer exports currently, the turbo::memory_map_remap() call
                    // would make the buffers invalid
                    return turbo::io_error("Cannot resize memory map while there are active readers");
                }

                if (new_size == 0) {
                    if (map_len_ > 0) {
                        // Just unmap the mmap and truncate the file to 0 size
                        region_.reset();
                        TURBO_RETURN_NOT_OK(::turbo::file_truncate(file_->fd(), 0));
                        map_len_ = offset_ = file_size_ = 0;
                    }
                    position_ = 0;
                    return turbo::OkStatus();
                }

                if (map_len_ > 0) {
                    void *result;
                    auto data = region_->data();
                    TURBO_RETURN_NOT_OK(turbo::memory_map_remap(data, map_len_, new_size,
                                                                     file_->fd(), &result));
                    region_->detach();  // avoid munmap() on destruction
                    region_ = std::make_shared<Region>(shared_from_this(),
                                                       static_cast<uint8_t *>(result), new_size);
                    map_len_ = file_size_ = new_size;
                    offset_ = 0;
                    if (position_ > map_len_) {
                        position_ = map_len_;
                    }
                } else {
                            DKCHECK_EQ(position_, 0);
                    // the mmap is not yet initialized, resize the underlying
                    // file, since it might have been 0-sized
                    TURBO_RETURN_NOT_OK(InitMMap(new_size, /*resize_file*/ true));
                }
                return turbo::OkStatus();
            }

            turbo::Status Seek(int64_t position) {
                if (position < 0) {
                    return turbo::invalid_argument_error("position is out of bounds");
                }
                position_ = position;
                return turbo::OkStatus();
            }

            turbo::Result<std::shared_ptr<Buffer>> slice(int64_t offset, int64_t length) {
                length = std::max<int64_t>(0, std::min(length, map_len_ - offset));

                if (length > 0) {
                            DKCHECK_NE(region_, nullptr);
                    return SliceBuffer(region_, offset, length);
                } else {
                    return std::make_shared<Buffer>(nullptr, 0);
                }
            }

            // map_len_ == file_size_ if memory mapping on the whole file
            int64_t size() const { return map_len_; }

            int64_t position() { return position_; }

            void advance(int64_t nbytes) { position_ = position_ + nbytes; }

            uint8_t *data() { return region_ ? region_->data() : nullptr; }

            uint8_t *head() { return data() + position_; }

            bool writable() { return file_->mode() != FileMode::READ; }

            bool opened() { return file_->is_open(); }

            int fd() const { return file_->fd(); }

            std::mutex &write_lock() { return file_->lock(); }

            std::mutex &resize_lock() { return resize_lock_; }

        private:
            // Initialize the mmap and set size, capacity and the data pointers
            turbo::Status InitMMap(int64_t initial_size, bool resize_file = false,
                            const int64_t offset = 0, const int64_t length = -1) {
                        DKCHECK(!region_);

                if (resize_file) {
                    TURBO_RETURN_NOT_OK(turbo::file_truncate(file_->fd(), initial_size));
                }

                int64_t mmap_length = initial_size;
                if (length >= 0) {
                    // memory mapping a file region
                    if (length > initial_size) {
                        return turbo::invalid_argument_error("mapping length is beyond file size");
                    }
                    mmap_length = length;
                }
                if (static_cast<int64_t>(static_cast<size_t>(mmap_length)) != mmap_length) {
                    return turbo::resource_exhausted_error("Requested memory map length ", mmap_length,
                                                 " does not fit in a C size_t "
                                                 "(are you using a 32-bit build of Nebula?)");
                }

                void *result = mmap(nullptr, static_cast<size_t>(mmap_length), prot_flags_, map_mode_,
                                    file_->fd(), static_cast<off_t>(offset));
                if (result == MAP_FAILED) {
                    return turbo::io_error("Memory mapping file failed: ",
                                           km_error(errno));
                }
                map_len_ = mmap_length;
                offset_ = offset;
                region_ = std::make_shared<Region>(shared_from_this(), static_cast<uint8_t *>(result),
                                                   map_len_);
                file_size_ = initial_size;

                return turbo::OkStatus();
            }

            std::unique_ptr<OSFile> file_;
            int prot_flags_;
            int map_mode_;

            std::shared_ptr<Region> region_;
            int64_t file_size_;
            int64_t position_;
            int64_t offset_;
            int64_t map_len_;
            std::mutex resize_lock_;
        };

        MemoryMappedFile::MemoryMappedFile() {}

        MemoryMappedFile::~MemoryMappedFile() { internal::CloseFromDestructor(this); }

        turbo::Result<std::shared_ptr<MemoryMappedFile>> MemoryMappedFile::create(
                const std::string &path, int64_t size) {
            TURBO_MOVE_OR_RAISE(auto file, FileOutputStream::open(path));
            TURBO_RETURN_NOT_OK(turbo::file_truncate(file->file_descriptor(), size));
            TURBO_RETURN_NOT_OK(file->close());
            return MemoryMappedFile::open(path, FileMode::READWRITE);
        }

        turbo::Result<std::shared_ptr<MemoryMappedFile>> MemoryMappedFile::open(const std::string &path,
                                                                         FileMode::type mode) {
            std::shared_ptr<MemoryMappedFile> result(new MemoryMappedFile());

            result->memory_map_.reset(new MemoryMap());
            TURBO_RETURN_NOT_OK(result->memory_map_->open(path, mode));
            return result;
        }

        turbo::Result<std::shared_ptr<MemoryMappedFile>> MemoryMappedFile::open(const std::string &path,
                                                                         FileMode::type mode,
                                                                         const int64_t offset,
                                                                         const int64_t length) {
            std::shared_ptr<MemoryMappedFile> result(new MemoryMappedFile());

            result->memory_map_.reset(new MemoryMap());
            TURBO_RETURN_NOT_OK(result->memory_map_->open(path, mode, offset, length));
            return result;
        }

        turbo::Result<int64_t> MemoryMappedFile::get_size() {
            TURBO_RETURN_NOT_OK(memory_map_->CheckClosed());
            return memory_map_->size();
        }

        turbo::Result<int64_t> MemoryMappedFile::tell() const {
            TURBO_RETURN_NOT_OK(memory_map_->CheckClosed());
            return memory_map_->position();
        }

        turbo::Status MemoryMappedFile::Seek(int64_t position) {
            TURBO_RETURN_NOT_OK(memory_map_->CheckClosed());
            return memory_map_->Seek(position);
        }

        turbo::Status MemoryMappedFile::close() { return memory_map_->close(); }

        bool MemoryMappedFile::closed() const { return memory_map_->closed(); }

        turbo::Result<std::shared_ptr<Buffer>> MemoryMappedFile::read_at(int64_t position,
                                                                 int64_t nbytes) {
            TURBO_RETURN_NOT_OK(memory_map_->CheckClosed());
            // if the file is writable, we acquire the lock before creating any slices
            // in case a resize is triggered concurrently, otherwise we wouldn't detect
            // a change in the use count
            auto guard_resize = memory_map_->writable()
                                ? std::unique_lock<std::mutex>(memory_map_->resize_lock())
                                : std::unique_lock<std::mutex>();

            TURBO_MOVE_OR_RAISE(
                    nbytes, internal::ValidateReadRange(position, nbytes, memory_map_->size()));
            // Arrange to page data in
            TURBO_RETURN_NOT_OK(turbo::memory_advise_will_need(
                    {{memory_map_->data() + position, static_cast<size_t>(nbytes)}}));
            return memory_map_->slice(position, nbytes);
        }

        turbo::Result<int64_t> MemoryMappedFile::read_at(int64_t position, int64_t nbytes, void *out) {
            TURBO_RETURN_NOT_OK(memory_map_->CheckClosed());
            auto guard_resize = memory_map_->writable()
                                ? std::unique_lock<std::mutex>(memory_map_->resize_lock())
                                : std::unique_lock<std::mutex>();

            TURBO_MOVE_OR_RAISE(
                    nbytes, internal::ValidateReadRange(position, nbytes, memory_map_->size()));
            if (nbytes > 0) {
                memcpy(out, memory_map_->data() + position, static_cast<size_t>(nbytes));
            }
            return nbytes;
        }

        turbo::Result<int64_t> MemoryMappedFile::read(int64_t nbytes, void *out) {
            TURBO_RETURN_NOT_OK(memory_map_->CheckClosed());
            TURBO_MOVE_OR_RAISE(int64_t bytes_read, read_at(memory_map_->position(), nbytes, out));
            memory_map_->advance(bytes_read);
            return bytes_read;
        }

        turbo::Result<std::shared_ptr<Buffer>> MemoryMappedFile::read(int64_t nbytes) {
            TURBO_RETURN_NOT_OK(memory_map_->CheckClosed());
            TURBO_MOVE_OR_RAISE(auto buffer, read_at(memory_map_->position(), nbytes));
            memory_map_->advance(buffer->size());
            return buffer;
        }

        Future<std::shared_ptr<Buffer>> MemoryMappedFile::read_async(const IOContext &,
                                                                    int64_t position,
                                                                    int64_t nbytes) {
            return Future<std::shared_ptr<Buffer>>::make_finished(read_at(position, nbytes));
        }

        turbo::Status MemoryMappedFile::will_need(const std::vector<ReadRange> &ranges) {

            TURBO_RETURN_NOT_OK(memory_map_->CheckClosed());
            auto guard_resize = memory_map_->writable()
                                ? std::unique_lock<std::mutex>(memory_map_->resize_lock())
                                : std::unique_lock<std::mutex>();

            std::vector<turbo::MemoryRegion> regions(ranges.size());
            for (size_t i = 0; i < ranges.size(); ++i) {
                const auto &range = ranges[i];
                TURBO_MOVE_OR_RAISE(
                        auto size,
                        internal::ValidateReadRange(range.offset, range.length, memory_map_->size()));
                        DKCHECK_NE(memory_map_->data(), nullptr);
                regions[i] = {const_cast<uint8_t *>(memory_map_->data() + range.offset),
                              static_cast<size_t>(size)};
            }
            return turbo::memory_advise_will_need(regions);
        }

        bool MemoryMappedFile::supports_zero_copy() const { return true; }

        turbo::Status MemoryMappedFile::write_at(int64_t position, const void *data, int64_t nbytes) {
            TURBO_RETURN_NOT_OK(memory_map_->CheckClosed());
            std::lock_guard<std::mutex> guard(memory_map_->write_lock());

            if (!memory_map_->opened() || !memory_map_->writable()) {
                return turbo::io_error("Unable to write");
            }
            TURBO_RETURN_NOT_OK(internal::ValidateWriteRange(position, nbytes, memory_map_->size()));

            TURBO_RETURN_NOT_OK(memory_map_->Seek(position));
            return WriteInternal(data, nbytes);
        }

        turbo::Status MemoryMappedFile::write(const void *data, int64_t nbytes) {
            TURBO_RETURN_NOT_OK(memory_map_->CheckClosed());
            std::lock_guard<std::mutex> guard(memory_map_->write_lock());

            if (!memory_map_->opened() || !memory_map_->writable()) {
                return turbo::io_error("Unable to write");
            }
            TURBO_RETURN_NOT_OK(
                    internal::ValidateWriteRange(memory_map_->position(), nbytes, memory_map_->size()));

            return WriteInternal(data, nbytes);
        }

        turbo::Status MemoryMappedFile::WriteInternal(const void *data, int64_t nbytes) {
            memcpy(memory_map_->head(), data, static_cast<size_t>(nbytes));
            memory_map_->advance(nbytes);
            return turbo::OkStatus();
        }

        turbo::Status MemoryMappedFile::resize(int64_t new_size) {
            TURBO_RETURN_NOT_OK(memory_map_->CheckClosed());
            std::unique_lock<std::mutex> write_guard(memory_map_->write_lock(), std::defer_lock);
            std::unique_lock<std::mutex> resize_guard(memory_map_->resize_lock(), std::defer_lock);
            std::lock(write_guard, resize_guard);
            TURBO_RETURN_NOT_OK(memory_map_->resize(new_size));
            return turbo::OkStatus();
        }

        int MemoryMappedFile::file_descriptor() const { return memory_map_->fd(); }

    }  // namespace io
}  // namespace nebula
